package com.pig4cloud.pig.consumer.task;

import com.alibaba.fastjson.JSONObject;
import com.pig4cloud.pig.admin.api.entity.NewsList;
import com.pig4cloud.pig.common.core.util.R;
import com.pig4cloud.pig.consumer.mapper.NewsListMapper;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
@Service
public class NewsConsumerTask {


	@Autowired
	private StringRedisTemplate redisTemplate;
	@Autowired
	private NewsListMapper newsListMapper;
	@Autowired
	private RestHighLevelClient client;
	@Autowired
	private ObjectMapper objectMapper;


	public static final String NEWS_LIST_KEY = "news_collection";


	@Scheduled(initialDelay = 1000, fixedDelay = 60000)
	public void consumeForList() {
		// 获取 List 的长度
		Long size = redisTemplate.opsForList().size(NEWS_LIST_KEY);
		log.info("### 开始搂取数据，数据大小为 : {}", size);
		if (size == null || size == 0) {
			return;
		}
		try {
			// 从 List 中取出所有数据（从 0 到 size-1）
			//List<String> rawList = redisTemplate.opsForList().range(NEWS_LIST_KEY, 0, size - 1);
			List<NewsList> newsItems = new ArrayList<>();
			Integer illeCount = 0;
			/*for (String item : rawList) {
				try {
					NewsList newsList = JSONObject.parseObject(item, NewsList.class);
					// 类型转换
					newsItems.add(newsList);
				}catch (Exception e) {
					e.printStackTrace();
					illeCount++;
				}
			}*/
			for(int i = 0 ; i < size ; i++) {
				String item = redisTemplate.opsForList().leftPop(NEWS_LIST_KEY);
				try {
					NewsList newsList = JSONObject.parseObject(item, NewsList.class);
					// 类型转换
					newsItems.add(newsList);
				}catch (Exception e) {
					// e.printStackTrace();
					log.info("news formatter failed : {}", e.getMessage());
					illeCount++;
				}
			}
			Integer affectedRows = newsListMapper.batchNewsInsert(newsItems);
			bulkInsert(newsItems);
			log.info("Inserted {} rows, skipped {} duplicates , error count {} ", affectedRows, newsItems.size() - affectedRows, illeCount);
		}catch (Exception e){
			redisTemplate.delete(NEWS_LIST_KEY);
			log.info("news formatter failed : {}", e.getMessage());
		}
	}

	public String insertData(NewsList newsList) throws IOException {
		IndexRequest request = new IndexRequest("news_list_sharding")
				.id(String.valueOf(newsList.getId()))
				.source(
						"title", newsList.getTitle(),
						"url", newsList.getUrl(),
						"create_time", newsList.getCreateTime()
				); // 或者使用 JSON 序列化

		IndexResponse response = client.index(request, RequestOptions.DEFAULT);
		return response.getResult().name();
	}

	// 批量插入数据
	public String bulkInsert(List<NewsList> newsLists) throws IOException {
		BulkRequest bulkRequest = new BulkRequest();
		for (NewsList news : newsLists) {
			bulkRequest.add(new IndexRequest("news_list_sharding")
					.id(String.valueOf(news.getId()))
					.source(objectMapper.writeValueAsString(news), XContentType.JSON)
			);
		}

		BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
		return response.hasFailures() ? "Failed" : "Success";
	}

}
